package org.zn.etl.service.impl;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.datax.common.entity.TableInfo;
import org.zn.etl.datasource.DataSourceFactoryHolder;
import org.zn.etl.datasource.entity.DataSourceInfo;
import org.zn.etl.jdbc.QueryHelper;
import org.zn.etl.job.dao.DataMappingMapper;
import org.zn.etl.job.dao.DataSourceInfoMapper;
import org.zn.etl.job.dao.JobGroupMapper;
import org.zn.etl.job.entity.ColumnMappingForm;
import org.zn.etl.job.entity.DataMapping;
import org.zn.etl.job.entity.EtlJob;
import org.zn.etl.job.entity.JobAnalyseForm;
import org.zn.etl.job.entity.JobGroup;
import org.zn.etl.job.entity.po.DataMappingPo;
import org.zn.etl.job.entity.po.DataSourceInfoPo;
import org.zn.etl.misc.BeanCopierUtils;
import org.zn.etl.misc.DataSourceInfo2ConfigMap;
import org.zn.etl.misc.DocHtmlBuilder;
import org.zn.etl.service.EtlAnalyseService;
import org.zn.etl.service.EtlPersistService;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class EtlAnalyseServiceImpl implements EtlAnalyseService {
	@Autowired
	private DataSourceFactoryHolder factoryHolder ; 
	
	@Autowired
	@Lazy
	private EtlPersistService etlPersistService;
	
	@Autowired
	private DataMappingMapper dataMappingMapper;
	
	@Autowired
	private DataSourceInfoMapper  dataSourceMapper;
	
	@Autowired
	private JobGroupMapper  jobGroupMapper;
	
	
	public void tableColumnMapping(EtlJob job,boolean merge) throws Exception {
		
		DataSourceInfo source=job.getSource();
		DataSourceInfo target=job.getTarget();
		
		Map<String, String> sourceConfig=DataSourceInfo2ConfigMap.toMap(source);

		Map<String, String> targetConfig=DataSourceInfo2ConfigMap.toMap(target);
		
		DataSource sSource=factoryHolder.getFactory(source.getFactoryType()).getDataSource(source.getKeyName(), sourceConfig);
		
		DataSource tSource=factoryHolder.getFactory(target.getFactoryType()).getDataSource(target.getKeyName(), targetConfig);
		
		List<DataMapping>  dmlist=	job.getDataMappings();
		
		for (DataMapping dataMapping : dmlist) {
			
			try {
			String sql=null;
			Map<String, ColumnInfo> scMap=Maps.newHashMap();
			Map<String, ColumnInfo> tcMap=Maps.newHashMap();
			
			String smeta=dataMapping.getSourceTable().getMetaTable();
			
			if(StringUtils.isEmpty(dataMapping.getSourceTable().getMetaTable())) {
				smeta=dataMapping.getSourceTable().getTable();
			}
			
			sql="select * from "+dataMapping.getSourceTable().getSchema()+"."+smeta  +" limit 1";
			
			log.info("execute  table:{} ,sql:{} ",dataMapping.getSourceTable().getTable(),sql);
			
			scMap=QueryHelper.queryMeta(sSource, sql);
			
			String tmeta=dataMapping.getTargetTable().getMetaTable();
			
			if(StringUtils.isEmpty(dataMapping.getTargetTable().getMetaTable())) {
				tmeta=dataMapping.getTargetTable().getTable();
			}

			sql="select * from "+dataMapping.getTargetTable().getSchema()+"."+tmeta+" limit 1";
			log.info("execute  table:{} ,sql:{} ",dataMapping.getTargetTable().getTable(),sql);
		
			tcMap=QueryHelper.queryMeta(tSource, sql);
			
			analyseColumn(scMap, tcMap, dataMapping,merge);
			
			} catch (Exception e) {
				e.printStackTrace();
			}
			
		}
		 

	}
	
	public void analyseColumn(Map<String, ColumnInfo> scMap,Map<String, ColumnInfo> tcMap,DataMapping dm,boolean merge) {
		
		List<Pair<ColumnInfo, ColumnInfo>> cmlist=dm.getColumnMappingPairs();

		List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs =new ArrayList<>();
		
		for (Entry<String,ColumnInfo>   entry : tcMap.entrySet()) {
			
			ColumnInfo  sci=scMap.remove(entry.getKey());
			
			columnMappingPairs.add(Pair.of(sci, entry.getValue()));
			
		}
		
		for (Entry<String,ColumnInfo>   entry : scMap.entrySet()) {
			
			ColumnInfo  tci=tcMap.get(entry.getKey());
			
			columnMappingPairs.add(Pair.of(entry.getValue(),tci));
			
		}
		
		if(CollectionUtils.isNotEmpty(cmlist)) {
		cmlist.addAll(columnMappingPairs);
		}else {
			dm.setColumnMappingPairs(columnMappingPairs);
		}
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public Integer analyseJob(JobAnalyseForm form) throws Exception {
		// TODO Auto-generated method stub
		
		EtlJob  job=etlPersistService.queryEtlJob(form.getJobName());
		
		tableColumnMapping(job,form.isMerge());
		
		DataMapping dm=job.getDataMappings().get(0);

		List<Pair<ColumnInfo, ColumnInfo>>  colsList= dm.getColumnMappingPairs();
		
		 List<ColumnInfo>  sourceColumns=new ArrayList<>();
		 List<ColumnInfo>  targetColumns=new ArrayList<>();
		
		ColumnMappingForm cmf=new ColumnMappingForm();
		cmf.setDataMappingId(dm.getId());
		cmf.setSourceColumns(sourceColumns);
		cmf.setTargetColumns(targetColumns);

		for (Pair<ColumnInfo, ColumnInfo> pair : colsList) {
				
			boolean flag=true;
			
			if("ignore".equals(form.getMissMatchColumn())) {
				flag= Objects.nonNull( pair.getLeft())&& Objects.nonNull(pair.getRight());
			}
			
			if(flag) {
			sourceColumns.add(pair.getLeft());
			targetColumns.add(pair.getRight());
			}
			
		}
	
		if(Boolean.TRUE.equals(form.isMerge())) {
		etlPersistService.clearColumnMappingByDmId(dm.getId());
		}
		
		etlPersistService.addDataMappingColumn(cmf);

		return colsList.size();
	}

	
	
	public String analyseTableRef(String srcds, String tagds, String schema,String tables) throws Exception {
		// TODO Auto-generated method stub
		EtlJob job=new EtlJob();
		DataSourceInfoPo sds=dataSourceMapper.selectByName(srcds);
		DataSourceInfoPo tds=dataSourceMapper.selectByName(tagds);
		
		
		DataSourceInfo  sdi=	BeanCopierUtils.copy(sds, new DataSourceInfo()); 
		DataSourceInfo  tdi=	BeanCopierUtils.copy(tds, new DataSourceInfo()); 
		
		job.setSource(sdi);
		job.setTarget(tdi);
		
		List<DataMappingPo> dmlist=	dataMappingMapper.tableRef(schema,tables);

	
		job.setDataMappings(dmlist.stream().map(it->{
			
			DataMapping dm=new DataMapping();
			
			TableInfo tt=new TableInfo();
			
			tt.setTable(it.getTargetTable());
			tt.setSchema(it.getTargetSchema());
			dm.setTargetTable(tt);
			
			TableInfo st=new TableInfo();
			st.setTable(it.getSourceTable());
			dm.setSourceTable(st);

			return dm;
			
		}).collect(Collectors.toList()));
		
		tableColumnMapping(job,false);
		
		String resHtml=  DocHtmlBuilder.wappedBody( DocHtmlBuilder.bulidTables(job),job.getJobName());

		return resHtml;
	}
	
	
	
	
	public String jobGroupDoc(String groupName) {
		
		JobGroup  jg=etlPersistService.queryJobGroupBy(groupName);
		
		List<EtlJob> jobs=jg.getJobList();
		
		List<EtlJob> fullJobs=new ArrayList<>();
		
		for (EtlJob etlJob : jobs) {
				
			EtlJob  fjob= etlPersistService.queryEtlJob(etlJob.getJobName());
			
			try {
				analyseDiff(fjob);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			fullJobs.add(fjob);
			
		}
		
		String  doc=DocHtmlBuilder.bulidJobListDoc(fullJobs,groupName);
		
		return doc;
		
	}
	
	
	private void analyseDiff(EtlJob  fjob) throws Exception {
		
		DataSourceInfo source=fjob.getSource();
		DataSourceInfo target=fjob.getTarget();
		
		Map<String, String> sourceConfig=DataSourceInfo2ConfigMap.toMap(source);

		Map<String, String> targetConfig=DataSourceInfo2ConfigMap.toMap(target);
		
		DataSource sSource=factoryHolder.getFactory(source.getFactoryType()).getDataSource(source.getKeyName(), sourceConfig);
		DataSource tSource=factoryHolder.getFactory(target.getFactoryType()).getDataSource(target.getKeyName(), targetConfig);
	
		 DataMapping dataMapping = fjob.getDataMappings().get(0);
		
		try {
			String sql=null;
			Map<String, ColumnInfo> scMap=Maps.newHashMap();
			Map<String, ColumnInfo> tcMap=Maps.newHashMap();
			
			String sourceTable=dataMapping.getSourceTable().getTable();
			if(StringUtils.isNotEmpty(dataMapping.getSourceTable().getMetaTable())) {
				sourceTable=dataMapping.getSourceTable().getMetaTable();
			}
			sql="select * from "+sourceTable+" where rownum=1";			
			log.info("execute  table:{} ,sql:{} ",dataMapping.getSourceTable().getTable(),sql);
			scMap=QueryHelper.queryMeta(sSource, sql);
			
			
			String targetTable=dataMapping.getTargetTable().getTable();
			
			if(StringUtils.isNotEmpty(dataMapping.getTargetTable().getMetaTable())) {
				targetTable=dataMapping.getTargetTable().getMetaTable();
			}
			sql="select * from "+dataMapping.getTargetTable().getSchema()+"."+targetTable+" limit 1";
			log.info("execute  table:{} ,sql:{} ",dataMapping.getTargetTable().getTable(),sql);
		
			tcMap=QueryHelper.queryMeta(tSource, sql);
			
			
			DataMapping orig=BeanCopierUtils.copy(dataMapping, new DataMapping());
			orig.setColumnMappingPairs(Lists.newArrayList());
			
			analyseColumn(scMap, tcMap, orig,false);
			
			mergeDiff(dataMapping.getColumnMappingPairs(), orig.getColumnMappingPairs());
			
			dataMapping.setColumnMappingPairs(orig.getColumnMappingPairs());
			
			} catch (Exception e) {
				e.printStackTrace();
			}

	}
	
	
	private void mergeDiff(List<Pair<ColumnInfo, ColumnInfo>>  setting,List<Pair<ColumnInfo, ColumnInfo>>  orig) {
		
		Map<String, Pair<ColumnInfo, ColumnInfo>>  rMap= setting.stream().collect(Collectors.toMap( it->{	 
			
			 //String l =it.getLeft()!=null? it.getLeft().getName():"";
			 String r=it.getRight()!=null? it.getRight().getName():"";
			
			return r;	
		 }
		 ,Function.identity(), (o,n)-> n , LinkedHashMap::new));
		
		Set<String> lName=new HashSet<>();
		
		for (int i = 0; i < orig.size(); i++) {
		
			Pair<ColumnInfo, ColumnInfo> pair=orig.get(i);
			
			if(  (pair.getRight()!=null && StringUtils.isNotEmpty(pair.getRight().getName())) ) {
				Pair<ColumnInfo, ColumnInfo>   sP= rMap.get(pair.getRight().getName());	
				if(sP!=null) {
					orig.set(i, sP);
					lName.add(sP.getLeft().getName());					
				}
			}
		}
		
		orig.removeIf(it->  it.getRight()==null && lName.contains(it.getLeft()!=null?it.getLeft().getName():""));
		
		
		
	}

	
	

}
